package com.hrsaas.ext.spider.core;

import java.net.ConnectException;
import java.net.URLDecoder;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import com.hrsaas.ext.spider.core.HttpClientBuilder.ChannelOptionSetting;
import com.hrsaas.ext.spider.core.MessageHandlerImpl.ResponseState;
import com.hrsaas.ext.spider.core.header.UserAgent;
import com.hrsaas.ext.spider.core.url.HostAndPort;
import com.hrsaas.ext.spider.core.url.URL;
import com.hrsaas.ext.spider.err.Exceptions;
import com.hrsaas.ext.spider.util.Assert;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.DefaultHttpContent;
import io.netty.handler.codec.http.DefaultHttpRequest;
import io.netty.handler.codec.http.DefaultLastHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.ssl.SslContext;
import io.netty.resolver.AddressResolverGroup;
import io.netty.util.AttributeKey;
import io.netty.util.IllegalReferenceCountException;

/**
 * <p>
 * 功能：http client,若多个请求是同属域名，默认会对响应的header及cookie进行处理，通过 isolated 关闭默认处理
 * <p>
 * @author wangxiaoliang@kanzhun.com
 *
 * 2018年8月2日
 */
public final class HttpClient {

	private final NioEventLoopGroup group;
	final boolean compress;
	private final int maxInitialLineLength;
	private final int maxChunkSize;
	private final int maxHeadersSize;
	private final boolean followRedirects;
	private final CharSequence userAgent;
	private final List<RequestInterceptor> interceptors;
	private final Iterable<ChannelOptionSetting<?>> settings;
	// common share cookie
	private static final CookieStore shareCookie = new CookieStore();
	private final Duration timeout;
	private final Timer timer = new Timer("HttpClient timeout for HttpClient@" + System.identityHashCode(this));
	private final SslBootstrapCache sslBootstraps;
	private final MessageHandlerImpl handler;
	private final AddressResolverGroup<?> resolver;
	private final NioChannelFactory channelFactory = new NioChannelFactory(Boolean.getBoolean("httpclient.debug"));

	public HttpClient() {
		this(false, 128 * 1024, 12, 8192, 16383, true, null, Collections.<RequestInterceptor>emptyList(),
				Collections.<ChannelOptionSetting<?>>emptyList(), null, null, null, null, null,-1);
	}

	/**
	 * 创建http client，构造函数中包含http client的所有可配置参数
	 *
	 * @param compress            启用http压缩
	 * @param maxChunkSize         buffer块大小
	 * @param threads              启动线程数
	 * @param maxInitialLineLength 最小初始化线程数
	 * @param maxHeadersSize       最大可支持的header数量
	 * @param followRedirects      是否允许重写向，可以为null，defualt true
	 * @param userAgent            useragent，可以为null,默认为chrome useragent
	 * @param interceptors         前置拦截器，可以为null
	 * @param settings             netty channel 参数，可以为null
	 * @param cookies              设置cookies，可以为null
	 * @param timeout              超时时间，可以为null
	 * @param sslContext           ssl设置 ，可以为null
	 * @param resolver             DNS 解析，可为null
	 * @param threadPool           netty线程池，可为null
	 * @param maxRedirects         最大接收重写向次数
	 */
	public HttpClient(boolean compress, int maxChunkSize, int threads, int maxInitialLineLength, int maxHeadersSize,
			boolean followRedirects, CharSequence userAgent, List<RequestInterceptor> interceptors,
			Iterable<ChannelOptionSetting<?>> settings, CookieStore cookies,
			Duration timeout, SslContext sslContext, AddressResolverGroup<?> resolver,
			NioEventLoopGroup threadPool, int maxRedirects) {
		group = threadPool == null ? new NioEventLoopGroup(threads, new TF()) : threadPool;
		this.compress = compress;
		this.resolver = resolver;
		this.maxInitialLineLength = maxInitialLineLength;
		this.maxChunkSize = maxChunkSize;
		this.maxHeadersSize = maxHeadersSize;
		this.followRedirects = followRedirects;
		this.userAgent = userAgent != null ? userAgent : UserAgent.getDefault();
		this.interceptors = interceptors == null ? Collections.<RequestInterceptor>emptyList()
				: new ImmutableList.Builder<RequestInterceptor>().addAll(interceptors).build();
		this.settings = settings == null ? Collections.<ChannelOptionSetting<?>>emptySet() : settings;
		shareCookie.addAll(cookies);
		this.timeout = timeout;
		this.handler = new MessageHandlerImpl(followRedirects, this, maxRedirects);
		sslBootstraps = new SslBootstrapCache(group, this.timeout, sslContext, this.handler, this.maxChunkSize,
				this.maxInitialLineLength, this.maxHeadersSize, this.compress, this.settings, resolver,
				channelFactory);
	}

	private static class TF implements ThreadFactory {

		private int threadsCreated = 0;

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(r, "HttpClient event loop " + ++threadsCreated);
			t.setDaemon(true);
			return t;
		}
	}

	/**
	 * 初始化 HttpClientBuilder，用于构建全局请求参数
	 * @return HttpClientBuilder
	 */
	public static HttpClientBuilder builder() {
		return new HttpClientBuilder();
	}

	public HttpClient shareCookie(CookieStore cookies) {
		shareCookie.addAll(cookies);
		return this;
	}
	
	/**
	 * 生成 get请求
	 *
	 * @return a request builder
	 */
	public HttpRequestBuilder get() {
		return new RB(HttpMethod.GET);
	}

	/**
	 * 生成 POST 请求
	 *
	 * @return a request builder
	 */
	public HttpRequestBuilder post() {
		return new RB(HttpMethod.POST);
	}

	private synchronized Bootstrap start(HostAndPort hostAndPort) {
		if (bootstrap == null) {
			bootstrap = new Bootstrap();
			if (resolver != null) {
				bootstrap.resolver(resolver);
			}
			bootstrap.group(group);
			bootstrap.handler(new Initializer(hostAndPort, handler, null, false, maxChunkSize,
					maxInitialLineLength, maxHeadersSize, compress));
			bootstrap.option(ChannelOption.TCP_NODELAY, true);
			bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
			bootstrap.option(ChannelOption.SO_REUSEADDR, false);
			if (timeout != null) {
				bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) timeout.toMillis());
			}
			for (ChannelOptionSetting<?> setting : settings) {
				setting.apply(bootstrap);
			}
			bootstrap.channelFactory(channelFactory);
		}
		return bootstrap;
	}

	private synchronized Bootstrap startSsl(HostAndPort hostAndPort) {
		return sslBootstraps.sslBootstrap(hostAndPort);
	}

	/**
	 * 关闭netty客户端
	 */
	@SuppressWarnings("deprecation")
	public void shutdown() {
		if (group != null) {
			group.shutdownGracefully(0, 10, TimeUnit.SECONDS);
			if (!group.isTerminated()) {
				group.shutdownNow();
			}
		}
		timer.cancel();
	}

	void copyHeaders(HttpRequest from, HttpRequest to, HttpHeaders headers, URL rurl) {
		for (Iterator<Map.Entry<CharSequence, CharSequence>> it = from.headers()
				.iteratorCharSequence(); it.hasNext();) {
			Map.Entry<CharSequence, CharSequence> e = it.next();
			to.headers().add(e.getKey(), e.getValue());
		}
		to.headers().set(HttpHeaderNames.HOST.toString(), rurl.getHost());
	}

	private void copyCookies(ResponseState res, RequestInfo info, HttpRequest nue) {
		CookieStore cookies = info.r.filterCookies(res.resp.headers());
		cookies.decorateAsync(nue);
		// redirect 时需要更新 response 的 host
		info.r.setDomain(nue.headers().get(HttpHeaderNames.HOST));
	}
	
	void redirect(HttpMethod method, URL url, RequestInfo info, 
			ResponseState res, ChannelHandlerContext ctx) {
		HttpRequest nue;
		if (method.equals(info.req.method())) {
			if (info.req instanceof DefaultFullHttpRequest) {
				DefaultFullHttpRequest dfrq = (DefaultFullHttpRequest) info.req;
				FullHttpRequest rq;
				try {
					rq = dfrq.copy();
				} catch (IllegalReferenceCountException e) { // Empty bytebuf
					rq = dfrq;
				}
				rq.setUri(url.getPath());
				nue = rq;
			} else {
				nue = new DefaultHttpRequest(info.req.protocolVersion(), info.req.method(),
						url.getPathAndQuery());
			}
		} else {
			nue = new DefaultHttpRequest(info.req.protocolVersion(), HttpMethod.valueOf(method.name()),
					url.getPath());
		}
		copyHeaders(info.req, nue, info.req.headers(), url);
		copyCookies(res, info, nue);
		submit(url, nue, info.cancelled, info.handle, info.r, info, info.remaining(), info.dontAggregate,
				info.chunkedBody);
	}

	private Bootstrap bootstrap;

	static final AttributeKey<RequestInfo> KEY = AttributeKey.<RequestInfo>valueOf("info");

	private final Set<ActivityMonitor> monitors = Sets.newConcurrentHashSet();

	public void addActivityMonitor(ActivityMonitor monitor) {
		monitors.add(monitor);
	}

	public void removeActivityMonitor(ActivityMonitor monitor) {
		monitors.remove(monitor);
	}

	private class AdapterCloseNotifier implements ChannelFutureListener {

		private final URL url;

		AdapterCloseNotifier(URL url) {
			this.url = url;
		}

		@Override
		public void operationComplete(ChannelFuture future) throws Exception {
			for (ActivityMonitor m : monitors) {
				m.onEndRequest(url);
			}
		}
	}

	static final class TimeoutTimerTask extends TimerTask implements ChannelFutureListener {

		private final AtomicBoolean cancelled;
		private final ResponseFuture handle;
		private final ResponseHandler<?> r;
		private final RequestInfo in;

		TimeoutTimerTask(AtomicBoolean cancelled, ResponseFuture handle, ResponseHandler<?> r, RequestInfo in) {
			Assert.notNull("in", in);
			Assert.notNull("cancelled", cancelled);
			this.cancelled = cancelled;
			this.handle = handle;
			this.r = r;
			this.in = in;
		}

		@Override
		public void run() {
			if (!cancelled.get()) {
				if (r != null) {
					r.onError(new TimeoutException(in.timeout.toString()));
				}
				if (handle != null) {
					handle.onTimeout(in.age());
				}
			}
			super.cancel();
		}

		@Override
		public void operationComplete(ChannelFuture f) throws Exception {
			cancelled.set(true);
			super.cancel();
		}
	}

	private void submit(final URL url, HttpRequest rq, final AtomicBoolean cancelled, final ResponseFuture handle,
			final ResponseHandler<?> r, RequestInfo info, Duration timeout, boolean noAggregate,
			final ChunkedContent chunked) {
		if (info != null && info.isExpired()) {
			// In case of a redirect, we may be called with an info that has
			// already expired if the timeout set in the builder has elapsed
			handle.event(new State.Timeout(info.age()));
			cancelled.set(true);
		}
		// Ensure the cancelled event is sent
		if (cancelled.get()) {
			handle.event(new State.Cancelled());
			return;
		}
		// Assign a reference to the channel as soon as it is available,
		// so that we can close it in case of an exception
		final AtomicReference<Channel> theChannel = new AtomicReference<>();
		// Allow any interceptors to amend or even replace the
		// request
		for (RequestInterceptor i : interceptors) {
			rq = i.intercept(rq);
		}
		final HttpRequest req = rq;
		try {
			HostAndPort hostAndPort = new HostAndPort(url.getHost(), url.getPort());
			Bootstrap bootstrap = url.getPort() != 443 ? start(hostAndPort)
                            : startSsl(hostAndPort);
			TimeoutTimerTask timerTask = null;
			boolean newRequest = info == null;
			if (info == null) {
				info = new RequestInfo(url, req, cancelled, handle, r, timeout, timerTask, noAggregate,
						chunked);
				if (timeout != null) {
					timerTask = new TimeoutTimerTask(cancelled, handle, r, info);
					timer.schedule(timerTask, timeout.toMillis());
				}
				info.timer = timerTask;
			}
			if (info.isExpired()) {
				// Check expiration again
				handle.event(new State.Timeout(info.age()));
				return;
			}
			handle.event(new State.Connecting());
			// XXX who is escaping this?
			req.setUri(req.uri().replaceAll("%5f", "_"));
			ChannelFuture fut = bootstrap.connect(url.getHost(), url.getPort());
			theChannel.set(fut.channel());
			if (timerTask != null) {
				fut.channel().closeFuture().addListener(timerTask);
			}
			fut.channel().attr(KEY).set(info);
			handle.setFuture(fut);
			if (!monitors.isEmpty()) {
				for (ActivityMonitor m : monitors) {
					m.onStartRequest(url);
				}
				fut.channel().closeFuture().addListener(new AdapterCloseNotifier(url));
			}
			if (newRequest && r != null) {
				handle.on(State.Error.class, new Receiver<Throwable>() {
					@Override
					public void receive(Throwable object) {
						r.onError(object);
					}
				});
				handle.on(StateType.Cancelled, new Receiver<Void>() {
					@Override
					public void receive(Void object) {
						r.onError(new CancellationException("Cancelled"));
					}
				});
				// set http client
				r.setClient(this);
			}

			fut.addListener(new ChannelFutureListener() {

				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if (!future.isSuccess()) {
						Throwable cause = future.cause();
						if (cause == null) {
							cause = new ConnectException(
									"Unknown problem connecting to " + url);
						}
						handle.event(new State.Error(cause));
	                if (r != null) {
	                    r.onError(cause);
	                }
						cancelled.set(true);
					}
					if (cancelled.get()) {
						future.cancel(true);
						if (future.channel().isOpen()) {
							future.channel().close();
						}
						for (ActivityMonitor m : monitors) {
							m.onEndRequest(url);
						}
						return;
					}
					handle.event(new State.Connected(future.channel()));
					handle.event(new State.SendRequest(req));
					future = future.channel().writeAndFlush(req);
					future.addListener(new ChannelFutureListener() {

						@Override
						public void operationComplete(final ChannelFuture future)
								throws Exception {
							if (cancelled.get()) {
								future.cancel(true);
								future.channel().close();
							}
							if (chunked != null) {
								handle.on(State.HeadersReceived.class,
										new Receiver<HttpResponse>() {
											boolean first = true;

											@Override
											public void receive(
													HttpResponse object) {
												if (first && (HttpResponseStatus.CONTINUE
														.equals(object.status()))) {
													first = false;
													ChannelFutureListener flusher = new ChannelFutureListener() {
														int count = 0;

														@Override
														public void operationComplete(
																ChannelFuture f)
																throws Exception {
															if (cancelled.get()) {
																if (f != null) {
																	f.cancel(true);
																	f.channel().close();
																	return;
																} else {
																	future.cancel(true);
																	future.channel().close();
																}
															}
															if (f != null && f
																	.cause() != null) {
																handle.event(new State.Error(
																		f.cause()));
																f.channel().close();
															}
															Channel ch = f == null
																	? future.channel()
																	: f.channel();
															Object chunk = chunked
																	.nextChunk(count++);
															if (chunk != null) {
																if (chunk instanceof ByteBuf) {
																	chunk = new DefaultHttpContent(
																			(ByteBuf) chunk);
																}
																ch.writeAndFlush(
																		chunk)
																		.addListener(this);
															} else {
																ch.writeAndFlush(
																		new DefaultLastHttpContent());
																handle.event(new State.AwaitingResponse());
															}
														}
													};
													try {
														flusher.operationComplete(
																null);
													} catch (Exception ex) {
														Exceptions.chuck(
																ex);
													}
												} else if (first) {
													if (isRedirect(object
															.status())) {
														handle.event(new State.AwaitingResponse());
													}
												}
											}
										});
							}
							handle.event(new State.AwaitingResponse());
						}
					});
				}

			});
		} catch (Exception ex) {
			Channel ch = theChannel.get();
			cancelled.set(true);
			if (ch != null && ch.isRegistered() && ch.isOpen()) {
				ch.close();
			}
			Exceptions.chuck(ex);
		}
	}

	private static final Set<HttpResponseStatus> REDIRECTS = new HashSet<>(
			Arrays.asList(HttpResponseStatus.FOUND, HttpResponseStatus.SEE_OTHER,
					HttpResponseStatus.TEMPORARY_REDIRECT, HttpResponseStatus.PERMANENT_REDIRECT));

	private static boolean isRedirect(HttpResponseStatus status) {
		return REDIRECTS.contains(status);
	}

	private static final class StoreHandler extends Receiver<HttpResponse> {

		private final CookieStore store;

		StoreHandler(CookieStore store) {
			this.store = store;
		}

		@Override
		public void receive(HttpResponse headerContainer) {
			store.extract(headerContainer.headers());
		}
	}

	private ByteBufAllocator alloc() {
		for (ChannelOptionSetting<?> setting : this.settings) {
			if (setting.option().equals(ChannelOption.ALLOCATOR)) {
				return (ByteBufAllocator) setting.value();
			}
		}
		return PooledByteBufAllocator.DEFAULT;
	}
	
	private final class RB extends RequestBuilder {

		RB(HttpMethod method) {
			super(method, alloc());
		}

		@Override
		public ResponseFuture execute(ResponseHandler<?> r) {
			URL u = getURL();
			HttpRequest req = build(u);
			if (userAgent != null) {
				req.headers().add(HttpHeaderNames.USER_AGENT, userAgent);
			}
			if (compress) {
				req.headers().add(HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP_DEFLATE);
			}
			AtomicBoolean cancelled = new AtomicBoolean();
			ResponseFuture handle = new ResponseFuture(cancelled);
			handle.handlers.addAll(super.handlers);
			handle.any.addAll(super.any);
			CookieStore theStore = HttpClient.shareCookie;
			// 多客户端共享cookie
			if (!theStore.isEmpty()) {
				HandlerEntry<? extends HttpResponse> entry = createHandler(State.HeadersReceived.class,
						new StoreHandler(theStore));
				handle.handlers.add(entry);
			}
			// transfer cookie  store and domain
			r.setCookieStore(super.store);
			r.setDomain(u.getHost());
			submit(u, req, cancelled, handle, r, null, this.timeout, noAggregate, chunkedContent());
			return handle;
		}

		private <T> HandlerEntry<T> createHandler(Class<? extends State<T>> event, Receiver<T> r) {
			HandlerEntry<T> result = new HandlerEntry<>(event);
			result.add(r);
			return result;
		}

		@Override
		@SuppressWarnings("unchecked")
		public <T> HttpRequestBuilder on(StateType event, Receiver<T> r) {
			super.on((Class<? extends State<T>>) event.type(), (Receiver<T>) event.wrapperReceiver(r));
			return this;
		}

	}
}
